﻿#include "SocketSystem.h"
#include "Log.h"
#include <event.h>
#include <event2/listener.h>
#include <event2/thread.h>
#include "Config.h"
#include <queue>
#include <condition_variable>
#include "cfg.h"

struct
{
	event_base* evb;
	WSADATA wsadata;
	std::unique_ptr<SocketMain> socket_main;
	std::unique_ptr<SocketSendThread> send_data_main;
	std::vector<std::shared_ptr<Channel>> channels;
	std::mutex channels_mutex;
	std::queue <std::pair<std::shared_ptr<Channel>, read_buffer_cache*>> be_send_datas;
	std::mutex send_data_lock;
	std::condition_variable send_data_cv;
	bool send_thread_close;

	int wsa_init;
	bool is_init;
}global{};

bool add_send_data(void* key, read_buffer_cache* buf)
{
	std::lock_guard<std::mutex> lg(global.send_data_lock);
	for(auto& item : global.channels)
	{
		if(item.get() == key)
		{
			global.be_send_datas.push({ item,buf });
			global.send_data_cv.notify_all();
			return true;
		}
	}
	return false;
}

void server_sub_read_cb(bufferevent* bev, void* ctx)
{
	read_buffer_cache* buf_ptr = new read_buffer_cache{};
	auto& buf = *buf_ptr;

	buf.size = bufferevent_read(bev, buf.data.data(), buf.data.size());
	if(!add_send_data(ctx, buf_ptr))
	{
		delete buf_ptr;
	}
}

constexpr timeval reconnect_tv{ 2,500 };

void server_sub_event_cb(struct bufferevent* bev, short what, void* ctx)
{
	Logger().warn("server_sub_event_cb {}", (void*)ctx);
	if(ctx != nullptr)
	{
		if(what & (BEV_EVENT_EOF | BEV_EVENT_ERROR))
		{
			TCPServerChannel* channel = (TCPServerChannel*)ctx;
			channel->del_client(bev);
		}
	}
}

void client_reconnect_cb(evutil_socket_t sock, short which, void* arg)
{
	Logger().warn("client_reconnect_cb {}", (void*)arg);
	if(arg != nullptr)
	{
		TCPClientChannel* channel = (TCPClientChannel*)arg;
		channel->reconnect();
	}
}

void client_event_cb(struct bufferevent* bev, short events, void* ctx)
{
	Logger().warn("client_event_cb {:b}", events);
	TCPClientChannel* channel = (TCPClientChannel*)ctx;
	if(events & (BEV_EVENT_ERROR))
	{
		channel->state = CHANNEL_STATE_UNCONNECT;
		emit channel->sig_state_changed(channel->state);
	}
	if(events & BEV_EVENT_EOF)
	{
		channel->state = CHANNEL_STATE_UNCONNECT;
		emit channel->sig_state_changed(channel->state);
	}
	if(events & BEV_EVENT_TIMEOUT)
	{
		channel->state = CHANNEL_STATE_TIMEOUT;
		emit channel->sig_state_changed(channel->state);
	}
	if(events & BEV_EVENT_CONNECTED)
	{
		channel->state = CHANNEL_STATE_CONNECTED;
		emit channel->sig_state_changed(channel->state);
		bufferevent_enable(bev, EV_READ);
	}
}

void listener_cb(evconnlistener* listener, evutil_socket_t fd, sockaddr* addr, int socklen, void* arg)
{
	if(arg != nullptr)
	{
		TCPServerChannel* channel = (TCPServerChannel*)arg;
		auto bev = bufferevent_socket_new(global.evb, fd, BEV_OPT_CLOSE_ON_FREE | BEV_OPT_THREADSAFE);
		bufferevent_setcb(bev, server_sub_read_cb, nullptr, server_sub_event_cb, arg);
		bufferevent_enable(bev, EV_READ);
		channel->add_client(bev);
	}

	Logger().info(u8"{} Link Me", fd);
}

void client_read_cb(bufferevent* bev, void* ctx)
{
	read_buffer_cache* buf_ptr = new read_buffer_cache{};
	auto& buf = *buf_ptr;

	buf.size = bufferevent_read(bev, buf.data.data(), buf.data.size());
	if(!add_send_data(ctx, buf_ptr))
	{
		delete buf_ptr;
	}
}

bool SocketSystemInitialize()
{
	if(global.is_init)
	{
		return global.is_init;
	}

	{
#ifdef _WINDOWS
		global.wsa_init = WSAStartup(WINSOCK_VERSION, &global.wsadata);
		if(global.wsa_init)
		{
			return false;
		}
		evthread_use_windows_threads();
#endif // _WINDOWS


		global.evb = event_base_new();

		////event_new(global.evb,evsignal_new)
		//global.evb_life_event = event_new(global.evb, -1, EV_PERSIST, [](evutil_socket_t, short, void*) { Logger().info("say hello"); }, nullptr);

		//event_add(global.evb_life_event, nullptr);



		/*SocketAddress sa{};
		sa.init("192.168.101.4", 1701);*/

		//UINT fg = LEV_OPT_CLOSE_ON_FREE | LEV_OPT_REUSEABLE;
		//auto listener = evconnlistener_new_bind(global.evb, listener_cb, nullptr, fg, 10, (sockaddr*)&addr, sizeof(sockaddr_in));

		global.socket_main = std::make_unique<SocketMain>();
		global.send_data_main = std::make_unique<SocketSendThread>();
		global.send_data_main->start();

		//global.socket_main->start();

		/*SocketAddress client_sa{};
		sa.init("127.0.0.1", 7002);

		SOCKET soc{ ~0ull };*/

		//bind(soc, client_sa.get(), client_sa.len());

		//soc = -1;

		//SocketAddress client_sa{};
		//sa.init("192.168.101.4", 7002);

		//intptr_t fd{ -1 };
		//auto soc = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);

		////bind(soc, client_sa.get(), client_sa.len());
		//connect(soc, sa.get(), sa.len());

		/*auto soc = socket(AF_INET, SOCK_STREAM, 0);

		SocketAddress client_sa{};
		client_sa.init_client(7001);

		bind(soc, client_sa.get(), client_sa.len());

		auto buffer_sock = bufferevent_socket_new(global.evb, soc, BEV_OPT_CLOSE_ON_FREE);

		bufferevent_setcb(buffer_sock, client_read_cb, nullptr, nullptr, nullptr);
		bufferevent_enable(buffer_sock, EV_READ);
		bufferevent_socket_connect(buffer_sock, sa.get(), sa.len());


		event_base_dispatch(global.evb);*/
	}

	global.is_init = true;
	return global.is_init;
}

void SocketSystemShutdown()
{
	global.send_thread_close = true;
	global.send_data_cv.notify_all();
	global.send_data_main->wait();

	global.send_data_main.reset();
	std::lock_guard lg{ global.channels_mutex };
	global.channels.clear();
	event_base_loopbreak(global.evb);
	global.socket_main->wait();
	event_base_free(global.evb);
	global.evb = nullptr;
	global.socket_main.reset();
}

Channel* CreateChannel(const ChannelCreateInfo* i)
{
	if(i == nullptr)
	{
		return nullptr;
	}
	std::lock_guard<std::mutex> lg{ global.channels_mutex };
	Channel* temp{};
	if(i->type & CHANNEL_PROTO_TYPE_TCP_SERVER)
	{
		auto c = std::make_shared<TCPServerChannel>(i);
		temp = c.get();
		global.channels.push_back(c);
	}
	if(i->type & CHANNEL_PROTO_TYPE_TCP_CLIENT)
	{
		auto c = std::make_shared<TCPClientChannel>(i);
		temp = c.get();
		global.channels.push_back(c);
	}
	if(i->type & CHANNEL_PROTO_TYPE_UDP)
	{
		auto c = std::make_shared<UDPChannel>(i);
		temp = c.get();
		global.channels.push_back(c);
	}
	return temp;
}

void DestoryChannel(Channel* c)
{
	if(c == nullptr)
	{
		return;
	}
	std::lock_guard<std::mutex> lg{ global.channels_mutex };
	for(auto& item : global.channels)
	{
		if(item.get() == c)
		{
			item = global.channels.back();
			global.channels.pop_back();
			break;
		}
	}
}

void SocketMain::run()
{
	Logger().info("SocketMain::run");
	if(global.evb != nullptr)
	{
		event_base_dispatch(global.evb);
	}
	Logger().info("SocketMain::End");
}

TCPServerChannel::TCPServerChannel(const ChannelCreateInfo* i) :Channel{ i }, listenner{}, clients{}
{
	//connect(this, &TCPServerChannel::sig_add_client, this, &TCPServerChannel::slot_add_client);
	if(info.type & CHANNEL_PROTO_TYPE_TCP_SERVER)
	{
		struct sockaddr* input_addr{};
		int addr_len{};
		if(info.input_address.valid())
		{
			input_addr = info.input_address.get();
			addr_len = info.input_address.len();
		}
		listenner = evconnlistener_new_bind(global.evb, listener_cb, this, LEV_OPT_CLOSE_ON_FREE | LEV_OPT_REUSEABLE, 10, input_addr, addr_len);
		if(!info.input_address.valid())
		{
			auto fd = evconnlistener_get_fd(listenner);
			auto len = info.input_address.len();
			getsockname(fd, info.input_address.get(), &len);
			info.input_address.sync();
		}
		this->state = CHANNEL_STATE_START;
		global.socket_main->start();
	}
}

TCPServerChannel::~TCPServerChannel()
{
	evconnlistener_disable(listenner);
	evconnlistener_free(listenner);
	for(auto& item : clients)
	{
		bufferevent_disable(item.first, EV_READ);
		bufferevent_free(item.first);
	}
}

TCPClientChannel::TCPClientChannel(const ChannelCreateInfo* i) :Channel{ i }, bev{}, reconnect_event{}
{
	if(info.type & CHANNEL_PROTO_TYPE_TCP_CLIENT)
	{

		intptr_t fd{ -1 };
		if(info.input_address.valid())
		{
			auto sock = socket(AF_INET, SOCK_STREAM, 0);
			if(INVALID_SOCKET == sock)
			{
				Logger().error("Create TCPClientChannel Error");
			}
			auto bind_res = bind(sock, info.input_address.get(), info.input_address.len());
			fd = sock;
		}
		bev = bufferevent_socket_new(global.evb, fd, BEV_OPT_CLOSE_ON_FREE | BEV_OPT_THREADSAFE);
		bufferevent_setcb(bev, client_read_cb, nullptr, client_event_cb, this);
		int res{ bufferevent_socket_connect(bev, info.output_address.get(), info.output_address.len()) };

		if(!info.input_address.valid())
		{
			int len = info.input_address.len();
			if(getsockname(bufferevent_getfd(bev), info.input_address.get(), &len))
			{
				Logger().warn("getsockname error");
			}
			info.input_address.sync();
		}

		auto enable_res = bufferevent_enable(bev, EV_READ);
		if(res == 0 && enable_res == 0)
		{
			this->state = CHANNEL_STATE_CONNECTED;
		}

		reconnect_event = event_new(global.evb, -1, EV_PERSIST, client_reconnect_cb, this);
		event_add(reconnect_event, &reconnect_tv);

		global.socket_main->start();
	}
}

TCPClientChannel::~TCPClientChannel()
{
	if(reconnect_event != nullptr)
	{
		event_del(reconnect_event);
		event_free(reconnect_event);
	}
	if(bev != nullptr)
	{
		bufferevent_enable(bev, EV_READ);
		bufferevent_free(bev);
	}
}

bool TCPClientChannel::reconnect()
{
	if(state == CHANNEL_STATE_UNCONNECT)
	{
		int res{ bufferevent_socket_connect(bev, info.output_address.get(), info.output_address.len()) };
		if(res < 0)
		{
			Logger().info("res < 0");
		}
	}
	return true;
}

void udp_read_cb(evutil_socket_t fd, short mod, void* arg)
{
	SocketAddress addr{};
	auto addr_size = addr.len();
	read_buffer_cache* buf_ptr = new read_buffer_cache{};
	auto& buf = *buf_ptr;
	buf.size = recvfrom(fd, buf.data.data(), buf.data.size(), 0, addr.get(), &addr_size);
	if(buf.size > 0)
	{
		Logger().info(buf.data.data());
		if(!add_send_data(arg, buf_ptr))
		{
			delete buf_ptr;
		}
	}
	else
	{
		Logger().warn("UDP Read Null Data");
		delete buf_ptr;
	}
}

UDPChannel::UDPChannel(const ChannelCreateInfo* i) :Channel{ i }, _socket{ INVALID_SOCKET }, udp_ev{}
{
	if(i->type & CHANNEL_PROTO_TYPE_UDP)
	{
		_socket = ::socket(AF_INET, SOCK_DGRAM, 0);
		auto bind_res = ::bind(_socket, info.input_address.get(), info.input_address.len());
		this->udp_ev = event_new(global.evb, _socket, EV_READ | EV_PERSIST, udp_read_cb, this);
		event_add(this->udp_ev, nullptr);
		if(_socket != INVALID_SOCKET && bind_res == 0 && udp_ev != nullptr)
		{
			this->state = CHANNEL_STATE_START;
		}
		global.socket_main->start();
	}
}

UDPChannel::~UDPChannel()
{
	if(this->udp_ev != nullptr)
	{
		event_del(this->udp_ev);
		event_free(this->udp_ev);
#ifdef _WIN32
		closesocket(this->_socket);
#endif // _WIN32
	}
}

void UDPChannel::SendDataImpl(read_buffer_cache& cache)
{
	if(state == CHANNEL_STATE_START)
	{
		for(auto& item : info.upd_address)
		{
			if(item.data_type == cache.data_type && item.data_type != invalid_value_id)
			{
				auto send_res = sendto(this->_socket, cache.data.data(), cache.size, 0, item.address.get(), item.address.len());
			}
		}
	}
}

void UDPChannel::start()
{
	if(state != CHANNEL_STATE_START)
	{
		_socket = ::socket(AF_INET, SOCK_DGRAM, 0);
		auto bind_res = ::bind(_socket, info.input_address.get(), info.input_address.len());
		this->udp_ev = event_new(global.evb, _socket, EV_READ | EV_PERSIST, udp_read_cb, this);
		event_add(this->udp_ev, nullptr);
		global.socket_main->start();
		state = CHANNEL_STATE_START;
		emit sig_state_changed(state);
	}
}

void UDPChannel::stop()
{
	if(state == CHANNEL_STATE_START)
	{
		state = CHANNEL_STATE_UNSTART;
		if(this->udp_ev != nullptr)
		{
			event_del(this->udp_ev);
			event_free(this->udp_ev);
#ifdef _WIN32
			closesocket(this->_socket);
#endif // _WIN32
			this->udp_ev = nullptr;
			this->_socket = INVALID_SOCKET;
		}
		emit sig_state_changed(state);
	}
}

void TCPServerChannel::SendDataImpl(read_buffer_cache& cache)
{
	if(cache.data_type == info.output_type && state == CHANNEL_STATE_START && info.output_type != invalid_value_id)
	{
		std::lock_guard<std::mutex> lg(this->clients_mutex);
		for(auto& item : this->clients)
		{
			SOCKET soc = bufferevent_getfd(item.first);
			auto send_res = send(soc, cache.data.data(), cache.size, 0);
		}
	}
}

void TCPClientChannel::SendDataImpl(read_buffer_cache& cache)
{
	if(cache.data_type == info.output_type && state == CHANNEL_STATE_CONNECTED && info.output_type != invalid_value_id)
	{
		SOCKET soc = bufferevent_getfd(this->bev);
		auto send_res = send(soc, cache.data.data(), cache.size, 0);
	}
}

void SocketSendThread::run()
{
	Logger().debug("Send Thread Begin");
	while(!global.send_thread_close)
	{
		std::unique_lock ulk{ global.send_data_lock };
		while((!global.send_thread_close) && global.be_send_datas.empty())
		{
			global.send_data_cv.wait(ulk);
		}

		if(global.send_thread_close)
		{
			return;
		}

		auto t = global.be_send_datas.front();
		global.be_send_datas.pop();
		ulk.unlock();

		global.channels_mutex.lock();
		auto all_channels = global.channels;
		global.channels_mutex.unlock();

		t.second->data_type = t.first->get_info().input_type;

		for(auto& item : all_channels)
		{
			item->SendDataImpl(*t.second);
		}

		delete t.second;
	}

	Logger().debug("Send Thread End");
}
